package cn.mesmile.flink.transform;

import cn.mesmile.flink.jdkstream.VideoOrder;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Date;

/**
 * @author zb
 * @date 2021/8/23 10:59
 * @Description
 */
public class FlinkTransformReduceApp {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStreamSource<VideoOrder> ds = env.fromElements(
                new VideoOrder("21312", "java", 32, 5, new Date()),
                new VideoOrder("314", "java", 32, 5, new Date()),
                new VideoOrder("542", "springboot", 32, 5, new Date()),
                new VideoOrder("42", "redis", 42, 5, new Date()),
                new VideoOrder("52", "java", 32, 5, new Date()),
                new VideoOrder("55", "test", 52, 5, new Date()),
                new VideoOrder("56", "cloud", 62, 5, new Date()),
                new VideoOrder("523", "redis", 42, 5, new Date())
                );

        KeyedStream<VideoOrder, String> videoOrderStringKeyedStream = ds.keyBy(new KeySelector<VideoOrder, String>() {
            @Override
            public String getKey(VideoOrder value) throws Exception {
                return value.getTitle();
            }
        });

        // value1是历史对象，value2是加入统计的对象，所以value1.f1是历史值，value2.f1是新值，不断累加
        DataStream<VideoOrder> reduce = videoOrderStringKeyedStream.reduce(new ReduceFunction<VideoOrder>() {
            @Override
            public VideoOrder reduce(VideoOrder value1, VideoOrder value2) throws Exception {
                VideoOrder videoOrder = new VideoOrder();
                videoOrder.setTitle(value1.getTitle());
                int money1 = value1.getMoney();
                int money2 = value2.getMoney();
                videoOrder.setMoney(money1 + money2);
                videoOrder.setCreateTime(new Date());
                return videoOrder;
            }
        });
        reduce.print("result：");

        env.execute("custom reduce app");
    }
}
